package com.cloudbroker.bcs.common.configure.zookeeper;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.cloudbroker.bcs.common.util.ConcurrentHashSet;
import com.cloudbroker.bcs.common.util.JSONUtil;
import com.cloudbroker.bcs.common.util.StringUtil;
import com.cloudbroker.bcs.common.zookeeper.NodeStringDataCacheListener;
import com.cloudbroker.bcs.common.zookeeper.ZooKeeperClient;

public class ZooKeeperConfigurator {
    
    private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperConfigurator.class);
    
    private static final String ZK_CONFIG_NOTIFIER_THREAD_NAME_PREFIX = "ZooKeeperConfigNotifier-";
    
    private static int zkConfigNotifierThreadIndex = 0;
    
    private ExecutorService zkConfigNotifierExecutor = Executors.newCachedThreadPool(new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r);
            t.setName(ZK_CONFIG_NOTIFIER_THREAD_NAME_PREFIX + (++zkConfigNotifierThreadIndex));
            t.setDaemon(true);
            return t;
        }
    });
    
    private Map<String, HierarchyPathWatcher> watchers = new HashMap<String, HierarchyPathWatcher>();
    
    private Lock watchersLock = new ReentrantLock();
    
    protected boolean compressData = false;
    
    protected ZooKeeperClient client;
    
    public ZooKeeperConfigurator(ZooKeeperClient client) {
        this.client = client;
    }
    
    public ZooKeeperClient getClient() {
        return client;
    }
    
    public void dispose() {
        for (HierarchyPathWatcher watcher : watchers.values()) {
            watcher.stop();
        }
        zkConfigNotifierExecutor.shutdown();
    }
    
    public boolean isCompressData() {
        return compressData;
    }
    
    public void setCompressData(boolean compressData) {
        this.compressData = compressData;
    }
    
    // /serviceGroupID/service/providers/
    
    // /serviceGroupID/configurations/serviceConfig/service/nodeID
    
    // /serviceGroupID/configurations/bizConfig/tenantID/queue/queueUri
    // /serviceGroupID/configurations/bizConfig/tenantID/agent/agentID
    
    // /brokers/brokerID/configurations/configCategory/
    // /serviceGroupID/configurations/bizConfig/brokerID/db/jdbc
    
    protected JSONObject getConfigOnPath(boolean noMerge, String fullPath) {
        JSONObject config = null;
        if (noMerge) {
            // 直接获取配置
            config = getConfigOnPath(fullPath);
        } else {
            // 将路径按分隔符拆散
            String[] pathSegs = StringUtil.splitUsingIndex(fullPath, ZooKeeperClient.PATH_SEPARATOR_CHAR);
            // 获取合并配置
            config = getMergeConfigOnPath(pathSegs);
        }
        return config;
    }
    
    protected JSONObject getConfigOnPath(boolean noMerge, String... pathSegs) {
        JSONObject config = null;
        if (noMerge) {
            // 直接构造完全的指定路径
            String fullPath = ZooKeeperClient.buildFullPath(pathSegs);
            // 获取配置
            config = getConfigOnPath(fullPath);
        } else {
            // 获取合并配置
            config = getMergeConfigOnPath(pathSegs);
        }
        return config;
    }
    
    protected JSONObject getMergeConfigOnPath(String... pathSegs) {
        JSONObject config = new JSONObject();
        if (null != pathSegs && 0 < pathSegs.length) {
            String path = String.valueOf(ZooKeeperClient.PATH_SEPARATOR_CHAR);
            JSONObject followingConfig = null;
            // 遍历路径段
            for (String pathSeg : pathSegs) {
                if (!StringUtils.isEmpty(pathSeg)) {
                    // 逐渐往后构建完整路径
                    path = ZooKeeperClient.appendPath(path, pathSeg);
                    // 获取在当前路径上的配置
                    followingConfig = getConfigOnPath(path);
                    if (LOG.isTraceEnabled()) {
                        LOG.trace("Config on path[" + path + "]: " + followingConfig);
                    }
                    // 合并配置
                    config = JSONUtil.merge(config, followingConfig);
                }
            }
        }
        return config;
    }
    
    protected JSONObject getConfigOnPath(String path) {
        String data = client.getStringData(path, compressData, false);
        if (StringUtils.isNotBlank(data)) {
            try {
                return JSON.parseObject(data);
            } catch (Exception e) {
                LOG.error("Invalid JSON, path: " + path + ", data: " + data, e);
            }
        }
        return new JSONObject();
    }
    
    public class ConfigReader {
        
        protected String root;
        
        protected String relativePath;
        
        protected String[] pathSegs;
        
        protected boolean noMerge;
        
        protected String[] concernFlags;
        
        protected String fullPath;
        
        public ConfigReader(String root, String... concernFlags) {
            if (StringUtils.isBlank(root)) {
                root = String.valueOf(ZooKeeperClient.PATH_SEPARATOR_CHAR);
            }
            this.root = root;
            this.concernFlags = concernFlags;
        }
        
        public ConfigReader onPath(String relativePath, String... concernFlags) {
            this.relativePath = relativePath;
            pathSegs = null;
            this.concernFlags = concernFlags;
            return this;
        }
        
        public ConfigReader onRoot() {
            this.relativePath = null;
            pathSegs = null;
            return this;
        }
        
        public ConfigReader noMerge() {
            this.noMerge = true;
            return this;
        }
        
        protected String getFullPath() {
            if (null == fullPath) {
                fullPath = ZooKeeperClient.buildFullPath(root, relativePath);
            }
            return fullPath;
        }
        
        public JSONObject getConfiguration() {
            if (noMerge) {
                return getConfigOnPath(true, getFullPath());
            } else {
                if (null == pathSegs) {
                    pathSegs = buildPathSegs();
                }
                return getConfigOnPath(false, pathSegs);
            }
        }
        
        protected String[] buildPathSegs() {
            if (null != relativePath) {
                String[] relativePathSegs = StringUtil.splitUsingIndex(relativePath,
                        ZooKeeperClient.PATH_SEPARATOR_CHAR);
                String[] pathSegs = new String[relativePathSegs.length + 1];
                pathSegs[0] = root;
                System.arraycopy(relativePathSegs, 0, pathSegs, 1, relativePathSegs.length);
                return pathSegs;
            } else {
                return new String[] { root };
            }
        }
        
        public void registerListener(ZooKeeperConfigurationListener listener) {
            ZooKeeperConfigurator.this.registerListener(root, relativePath, listener, concernFlags);
        }
        
        public void unregisterListener(ZooKeeperConfigurationListener listener) {
            ZooKeeperConfigurator.this.unregisterListener(getFullPath(), listener);
        }
        
        public String getRoot() {
            return root;
        }
        
    }
    
    protected void setConfigOnPath(boolean overwrite, String key, Object value, String fullPath) {
        if (!StringUtils.isBlank(fullPath)) {
            JSONObject config = null;
            if (!overwrite) {
                config = getConfigOnPath(fullPath);
            }
            config = JSONUtil.merge(config, key, value);
            client.setData(fullPath, config.toString(), compressData);
        }
    }
    
    protected void setConfigOnPath(boolean overwrite, String key, Object value, String... pathSegs) {
        String fullPath = ZooKeeperClient.buildFullPath(pathSegs);
        setConfigOnPath(overwrite, key, value, fullPath);
    }
    
    protected void setConfigOnPath(boolean overwrite, Map<String, Object> map, String fullPath) {
        if (!StringUtils.isBlank(fullPath)) {
            JSONObject config = null;
            if (!overwrite) {
                config = getConfigOnPath(fullPath);
            }
            config = JSONUtil.merge(config, map);
            client.setData(fullPath, config.toString(), compressData);
        }
    }
    
    protected void setConfigOnPath(boolean overwrite, Map<String, Object> map, String... pathSegs) {
        String fullPath = ZooKeeperClient.buildFullPath(pathSegs);
        setConfigOnPath(overwrite, map, fullPath);
    }
    
    protected void setConfigOnPath(boolean overwrite, Properties prop, String fullPath) {
        if (!StringUtils.isBlank(fullPath)) {
            JSONObject config = null;
            if (!overwrite) {
                config = getConfigOnPath(fullPath);
            }
            config = JSONUtil.merge(config, prop);
            client.setData(fullPath, config.toString(), compressData);
        }
    }
    
    protected void setConfigOnPath(boolean overwrite, Properties prop, String... pathSegs) {
        String fullPath = ZooKeeperClient.buildFullPath(pathSegs);
        setConfigOnPath(overwrite, prop, fullPath);
    }
    
    public class ConfigWriter {
        
        protected String root;
        
        protected String relativePath;
        
        protected boolean overwrite;
        
        protected String fullPath;
        
        public ConfigWriter(String root) {
            this.root = root;
        }
        
        public ConfigWriter onPath(String relativePath) {
            this.relativePath = relativePath;
            return this;
        }
        
        public ConfigWriter onRoot() {
            this.relativePath = null;
            return this;
        }
        
        public ConfigWriter overwrite() {
            this.overwrite = true;
            return this;
        }
        
        protected String getFullPath() {
            if (null == fullPath) {
                fullPath = ZooKeeperClient.buildFullPath(root, relativePath);
            }
            return fullPath;
        }
        
        public void setConfiguration(String key, Object value) {
            setConfigOnPath(overwrite, key, value, getFullPath());
        }
        
        public void setConfiguration(Map<String, Object> map) {
            setConfigOnPath(overwrite, map, getFullPath());
        }
        
        public void setConfiguration(Properties prop) {
            setConfigOnPath(overwrite, prop, getFullPath());
        }
        
        public String getRoot() {
            return root;
        }
        
    }
    
    protected class HierarchyPathWatcher extends NodeStringDataCacheListener {
        
        private String fullPath;
        
        private String[] concernFlags;
        
        private boolean started;
        
        private HierarchyPathWatcher parentWatcher;
        
        private Set<HierarchyPathWatcher> childWatchers = new ConcurrentHashSet<HierarchyPathWatcher>();
        
        private Set<ZooKeeperConfigurationListener> listeners = new ConcurrentHashSet<ZooKeeperConfigurationListener>();
        
        private JSONObject parentConfig;
        
        private JSONObject thisConfig;
        
        public HierarchyPathWatcher(String fullPath) {
            this.fullPath = fullPath;
            dataCompressed = compressData;
        }
        
        public void setConcernFlags(String... concernFlags) {
            this.concernFlags = concernFlags;
        }
        
        @Override
        public void nodeChanged(String path, String data) {
            // 解析字符串为JSON对象
            JSONObject configuration = parseStringDataToJSONObject(path, data);
            // 保存本节点的配置
            thisConfig = configuration;
            // 获取合并的配置
            JSONObject mergedConfig = getMergedConfigOnThisNode();
            // 通知本节点的监听器及子节点的观察者
            notify(mergedConfig);
        }
        
        private JSONObject parseStringDataToJSONObject(String path, String data) {
            if (StringUtils.isNotBlank(data)) {
                try {
                    return JSON.parseObject(data);
                } catch (Exception e) {
                    LOG.error("Invalid JSON, path: " + path + ", data: " + data, e);
                }
            }
            return new JSONObject();
        }
        
        public void notify(JSONObject configuration) {
            // 通知本节点的监听器
            notifyThisNode(configuration);
            // 通知子节点的观察者
            notifyChildWatchers(configuration);
        }
        
        public void notifyThisNode(final JSONObject configuration) {
            if (listeners.isEmpty()) {
                return;
            }
            zkConfigNotifierExecutor.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        for (ZooKeeperConfigurationListener listener : listeners) {
                            listener.configurationChanged(configuration, fullPath, concernFlags);
                        }
                    } catch (Exception e) {
                        LOG.error("notify configurationChanged failed", e);
                    }
                }
            });
        }
        
        public void notifyChildWatchers(final JSONObject configuration) {
            if (childWatchers.isEmpty()) {
                return;
            }
            zkConfigNotifierExecutor.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        for (HierarchyPathWatcher watcher : childWatchers) {
                            watcher.parentConfigChanged(fullPath, configuration);
                        }
                    } catch (Exception e) {
                        LOG.error("notify parentConfigChanged failed", e);
                    }
                }
            });
        }
        
        public void parentConfigChanged(String parentFullPath, JSONObject parentPathConfig) {
            // 保存父节点的配置
            parentConfig = parentPathConfig;
            // 获取合并的配置
            JSONObject mergedConfig = getMergedConfigOnThisNode();
            // 通知本节点的监听器及子节点的观察者
            notify(mergedConfig);
        }
        
        public JSONObject getNoMergeConfigOnThisNode() {
            if (null == thisConfig) {
                // 初始化本节点的配置
                String data = client.getStringData(fullPath, dataCompressed, false);
                thisConfig = parseStringDataToJSONObject(fullPath, data);
            }
            return thisConfig;
        }
        
        public JSONObject getParentConfig() {
            if (null == parentConfig) {
                // 初始化父节点的配置
                if (null == parentWatcher) {
                    parentConfig = new JSONObject();
                } else {
                    parentConfig = parentWatcher.getMergedConfigOnThisNode();
                }
            }
            // 必须clone返回，否则由于merge时会改变其内容，会造成并发的对其遍历读取的线程异常
            return JSONUtil.clone(parentConfig);
        }
        
        public JSONObject getMergedConfigOnThisNode() {
            return JSONUtil.merge(getParentConfig(), getNoMergeConfigOnThisNode());
        }
        
        public void start() {
            if (!started) {
                client.watchData(fullPath, this);
                started = true;
            }
        }
        
        public void stop() {
            if (started) {
                client.removeWatchDataListener(fullPath, this);
                started = false;
            }
        }
        
        public void addChildWatcher(HierarchyPathWatcher watcher) {
            childWatchers.add(watcher);
            watcher.setParentWatcher(this);
        }
        
        public void removeChildWatcher(HierarchyPathWatcher watcher) {
            childWatchers.remove(watcher);
        }
        
        public void addListener(ZooKeeperConfigurationListener listener) {
            if (listeners.add(listener)) {
                listener.configurationChanged(getMergedConfigOnThisNode(), fullPath, concernFlags);
            }
        }
        
        public void removeListener(ZooKeeperConfigurationListener listener) {
            listeners.remove(listener);
        }
        
        public String getFullPath() {
            return fullPath;
        }
        
        public void setParentWatcher(HierarchyPathWatcher parentWatcher) {
            this.parentWatcher = parentWatcher;
        }
        
        @Override
        public int hashCode() {
            final int prime = 31;
            int result = 1;
            result = prime * result + getOuterType().hashCode();
            result = prime * result + ((fullPath == null) ? 0 : fullPath.hashCode());
            return result;
        }
        
        @Override
        public boolean equals(Object obj) {
            if (this == obj) {
                return true;
            }
            if (obj == null) {
                return false;
            }
            if (!(obj instanceof HierarchyPathWatcher)) {
                return false;
            }
            HierarchyPathWatcher other = (HierarchyPathWatcher) obj;
            if (!getOuterType().equals(other.getOuterType())) {
                return false;
            }
            if (fullPath == null) {
                if (other.fullPath != null) {
                    return false;
                }
            } else if (!fullPath.equals(other.fullPath)) {
                return false;
            }
            return true;
        }
        
        private ZooKeeperConfigurator getOuterType() {
            return ZooKeeperConfigurator.this;
        }
        
        @Override
        public String toString() {
            return "HierarchyPathWatcher [fullPath=" + fullPath + "]";
        }
        
    }
    
    protected void registerListener(String root, String relativePath, ZooKeeperConfigurationListener listener,
            String... concernFlags) {
        HierarchyPathWatcher watcher = null;
        // 单独处理根节点的情况
        if (StringUtils.isEmpty(relativePath)) {
            watcher = getHierarchyPathWatcher(root);
            watcher.setConcernFlags(concernFlags);
            watcher.addListener(listener);
            watcher.start();
            return;
        }
        // 去除开头的路径分隔符
        relativePath = ZooKeeperClient.ensureChildPathRelative(relativePath);
        // 将指定路径按分隔符拆散
        String[] pathUnits = StringUtil.splitUsingIndex(relativePath, ZooKeeperClient.PATH_SEPARATOR_CHAR);
        // 创建相关观察者列表
        List<HierarchyPathWatcher> relatedWatchers = new ArrayList<HierarchyPathWatcher>(pathUnits.length + 1);
        // 父观察者从根节点开始
        HierarchyPathWatcher parentWatcher = getHierarchyPathWatcher(root);
        relatedWatchers.add(parentWatcher);
        String parentPath = root;
        String currentPath = null;
        for (String pathUnit : pathUnits) {
            // 逐层向后拼接路径
            currentPath = ZooKeeperClient.appendPath(parentPath, pathUnit);
            // 获取本层的观察者
            watcher = getHierarchyPathWatcher(currentPath);
            // 向父观察者注册本层观察者
            parentWatcher.addChildWatcher(watcher);
            // 添加本层观察者到相关观察者列表中
            relatedWatchers.add(watcher);
            // 更新父路径
            parentPath = currentPath;
            // 更新父观察者
            parentWatcher = watcher;
        }
        watcher.setConcernFlags(concernFlags);
        // 注册监听器
        watcher.addListener(listener);
        // 逐层启动观察者
        for (HierarchyPathWatcher relatedWatcher : relatedWatchers) {
            relatedWatcher.start();
        }
    }
    
    protected HierarchyPathWatcher getHierarchyPathWatcher(String fullPath) {
        HierarchyPathWatcher watcher = watchers.get(fullPath);
        if (null == watcher) {
            watchersLock.lock();
            try {
                watcher = watchers.get(fullPath);
                if (null == watcher) {
                    watcher = new HierarchyPathWatcher(fullPath);
                    watchers.put(fullPath, watcher);
                }
            } finally {
                watchersLock.unlock();
            }
        }
        return watcher;
    }
    
    protected void unregisterListener(String fullPath, ZooKeeperConfigurationListener listener) {
        HierarchyPathWatcher watcher = watchers.get(fullPath);
        if (null != watcher) {
            watcher.removeListener(listener);
        }
    }
    
}
